package com.seventeen.mq.proxy.server.proxy.ext;

import com.alibaba.fastjson.JSON;
import com.seventeen.mq.proxy.common.constant.AccessChannelConstants;
import com.seventeen.mq.proxy.common.exception.base.BaseException;
import com.seventeen.mq.proxy.common.utils.StringUtils;
import lombok.*;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 4.x 生产者代理实体类
 *
 * @author seventeen
 */
@EqualsAndHashCode(callSuper = true)
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DefaultExtMqProducer extends BaseExtMqProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultExtMqProducer.class);
    DefaultMQProducer mqProducer;

    private RPCHook getAclRpcHook(String accessKey, String secretKey) {
        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
    }

    public void init() {
        try {
            String accessKey = getAccessKey();
            String secretKey = getSecretKey();
            if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey)) {
                mqProducer = new DefaultMQProducer(getProducerProxyGroup(), getAclRpcHook(accessKey, secretKey));
            } else {
                mqProducer = new DefaultMQProducer(getProducerProxyGroup());
            }
            mqProducer.setNamesrvAddr(getEndpoints());
            if (AccessChannelConstants.CLOUD.equals(getAccessChannel())) {
                // 设置接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项；如果不开启消息轨迹功能，则不需要运行此项。
                mqProducer.setAccessChannel(AccessChannel.CLOUD);
            }
            mqProducer.start();
        } catch (Exception e) {
            LOGGER.error("DefaultExtMQProducer start fail:{},{}", getProducerProxyGroup(), getEndpoints(), e);
        }
    }

    public void destroy() {
        LOGGER.info("{}  shutdown", this);
        mqProducer.shutdown();
    }

    @Override
    public void sendMsg(ExtSendMessage sendMessage) {
        String topic = sendMessage.getTopic();
        String tag = sendMessage.getTag();
        String keys = sendMessage.getKeys();
        String message = sendMessage.getMessage();
        try {
            Message msg;
            if (StringUtils.isBlank(keys)) {
                msg = new Message(topic, tag, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
            } else {
                msg = new Message(topic, tag, keys, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
            }

            //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            //RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
            Integer delayTimeLevel = sendMessage.getDelayTimeLevel();
            if (sendMessage.getDelayTimeLevel() != null) {
                if (delayTimeLevel >= MIN_DELAY_TIME_LEVEL && delayTimeLevel <= MAX_DELAY_TIME_LEVEL) {
                    msg.setDelayTimeLevel(sendMessage.getDelayTimeLevel());
                } else {
                    LOGGER.error("send message error,delayTimeLevel must in 1->18,but {}", delayTimeLevel);
                    throw new BaseException(String.format("delayTimeLevel %s is bad value", delayTimeLevel));
                }
            }
            SendResult sendResult = mqProducer.send(msg);
            LOGGER.info("send message end, msg {},  sendResult {}", msg, JSON.toJSONString(sendResult));

        } catch (Exception e) {
            LOGGER.error("send message error, msg {} ,", JSON.toJSONString(sendMessage), e);
            throw new BaseException("消息发送失败");
        }
    }

    @Override
    public String toString() {
        return "DefaultExtMQProducer{" + "nameServerAddr='" + getEndpoints() + '\'' + ", producerProxyGroup='" + getProducerProxyGroup() + '\'' + ", version='" + getVersion() + '\'' + '}';
    }
}
